using System;
using System.Collections;
using System.Collections.Generic;

//using System.Diagnostics;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Security.Cryptography;
using System.Text;

//using System.Text.RegularExpressions;
using System.Threading;

namespace Calibur {
	public class Sender {
		private const int CRYPTO_KEY_LEN = 24;
		private const int SHA_LEN = 20;
		private const UInt64 MASK = 0x00000000000000ff;
		private Session session;
		private Queue<Message> queue;
		private MemoryStream buf;
		private TripleDESCryptoServiceProvider des3Provider;
		private SHA1CryptoServiceProvider sha1Provider;
		private byte[] cryptoKey;
		private UInt64 sequence;
		private byte[][] salt;
		private byte[] sha;
		private byte[] mixed;
		private UInt64 lastTimeStamp = 0;

		public UInt64 Sequence {
			get { return sequence; }
		}
		
		private void InitSalt() {
			salt = new byte[KeyData.fibonacci.Length][];
			byte[] bytes = new byte[sizeof(UInt64)];
			UInt64 v;
			for (int i = 0; i < KeyData.fibonacci.Length; ++i) {
				v = KeyData.fibonacci[i];
				for (int j = 0; j < sizeof(UInt64); ++j) {
					bytes[j] = (byte)((v >> (j << 3)) & MASK);
				}
				salt[i] = sha1Provider.ComputeHash(bytes);
			}
		}

		public Sender(Session session) {
			this.session = session;
			this.queue = new Queue<Message>();
			this.buf = new MemoryStream();
			/* 应用指令报文顺序号，！！！注意不包含心跳等控制指令 */
			this.sequence = 0;
			
			/* 初始化3DES加密provider */
			des3Provider = new TripleDESCryptoServiceProvider();
			des3Provider.Mode = CipherMode.ECB;
			des3Provider.Padding = PaddingMode.PKCS7;
			
			sha1Provider = new SHA1CryptoServiceProvider();
			cryptoKey = new byte[CRYPTO_KEY_LEN];
			mixed = new byte[sizeof(UInt64)];
			
			/* 初始化斐波那契SHA */
			InitSalt();
		}
		
		private void UpdateSHA(UInt64 cts, UInt64 sts) {
			if (lastTimeStamp == cts) {
				return;
			}
			
			byte[] segment = new byte[sizeof(UInt64)];
			int i, bits;
			byte b1, b2;
			for (i = 0; i < sizeof(UInt64); ++i) {
				bits = i << 3;
				b1 = (byte)((cts >> bits) & MASK);
				b2 = (byte)((sts >> bits) & MASK);
				segment[i] = (byte)(b1 ^ b2);
			}
			
			sha = sha1Provider.ComputeHash(segment);
			lastTimeStamp = cts;
		}
		
		public void CreateCryptoKey(UInt64 cts, UInt64 sts, UInt64 sequ, byte[] key) {
			UpdateSHA(cts, sts);
			
			int i, j, bits;
			UInt64 mv = KeyData.feeds[sequ % (UInt64) KeyData.feeds.Length];
			for (i = 0; i < sizeof(UInt64); ++i) {
				bits = i << 3;
				mixed[i] = (byte) ((mv >> bits) & MASK);
			}
			
			byte[] cur = salt[sequ % (UInt64)KeyData.fibonacci.Length];
			Array.Clear(key, 0, CRYPTO_KEY_LEN);
			
			Array.Copy(cur, key, SHA_LEN);
			for (i = 0; i < SHA_LEN; ++i) {
				key[i + 4] ^= sha[i];
			}
			
			for (i = 0, j = 0; i < CRYPTO_KEY_LEN; ++i, ++j) {
				if (j == 8) {
					j = 0;
				}
			
				key[i] ^= mixed[j];
			}
		}

		public void Send(Message message) {
			queue.Enqueue(message);
			lock (this) {
				Monitor.Pulse(this);
			}
		}
		
		public void Reset() {
			queue.Clear();
			buf.SetLength(0);
			buf.Seek(0, SeekOrigin.Begin);
			sequence = 0;
		}

		public void Send(MessageBean mbean) {
			CommandMessage message = new CommandMessage(mbean, session.MessageBeanFactory);
			Send(message);
		}
		
		private bool SendToNet(Message message) {
			Socket socket = session.Socket;
			Int32 msgSize = 0;
			buf.SetLength(0);
			
			/* 预留1个int（4个字节）的message size空间 */
			BinaryMessageBeanFormat.PutInt32(buf, msgSize);
			
			/* 1个字节的message type */
			byte messageType = (byte)(int)message.Type;
			if (message.Type == MessageType.App && session.Crypto) {
				messageType |= 0x80;
			}
			BinaryMessageBeanFormat.PutByte(buf, messageType);
			
			/* 消息体，如果messageType==App，则是MessageBean */
			if (message.Type == MessageType.App && session.Crypto) {
				CreateCryptoKey(session.ClientTimestamp, session.ServerTimestamp, sequence, cryptoKey);
				Console.WriteLine("RealKey ...");
				Console.WriteLine(Logger.ToHexString(cryptoKey));
				CryptoStream cryptoStream = new CryptoStream(buf,
                    des3Provider.CreateEncryptor(cryptoKey, null),
                    CryptoStreamMode.Write);
				message.Serialize(cryptoStream);
				cryptoStream.FlushFinalBlock();
			} else {
				message.Serialize(buf);
			}
			
			/* 回填message size */
			int total = (int)buf.Length;
			msgSize = total - 4;
			buf.Seek(0, SeekOrigin.Begin);
			BinaryMessageBeanFormat.PutInt32(buf, msgSize);
			
			/* 发送数据 */
			int remaining = total;
			int sent = 0;
			int cnt = 0;
			byte[] bytes = buf.GetBuffer();
			while (remaining > 0) {
				cnt = socket.Send(bytes, sent, remaining, SocketFlags.None);
				if (cnt <= 0) {
					return false;
				} else {
					remaining -= cnt;
					sent += cnt;
				}
			}
			
			return true;
		}

		public void ThreadRun() {
			while (true) {
				lock (this) {
					try {
						Monitor.Wait(this);
					} catch (Exception e) {
						session.onSenderException(e);
						return;
					}
				}

				while (queue.Count > 0) {
					try {
						Message message = queue.Dequeue();
						if (!SendToNet(message)) {
							session.onSenderException(new SessionLostException());
							return;
						}
						
						if (message.Type == MessageType.App) {
							++sequence;
						}
					} catch (Exception e) {
						session.onSenderException(e);
						return;
					}
				}
			}
		}
	}
	
	internal class MessageReader {
		enum RecvPhase {
			MessageSize,
			MessageBody
		}
		
		Socket socket;
		int remaining, got, toRead;
		RecvPhase phase;
		byte[] bytes;
		MemoryStream recvBuffer;
			
		public MessageReader(int remaining) {
			bytes = new byte[1024];
			recvBuffer = new MemoryStream();
			Reset(RecvPhase.MessageSize, remaining);
		}
			
		public MessageReader() : this(0) {
		}
		
		public void Reset(Socket socket) {
			this.socket = socket;
			Reset(RecvPhase.MessageSize, 4);
		}
			
		private void Reset(RecvPhase phase, int remaining) {
			this.phase = phase;
			this.remaining = remaining;
			this.got = 0;
			this.toRead = 0;
			
			recvBuffer.SetLength(0);
			recvBuffer.Seek(0, SeekOrigin.Begin);
		}
		
		private int socketRecv(byte[] buffer, int start, int count) {
			int recvBytes = socket.Receive(buffer, start, count, SocketFlags.None);
			if (recvBytes <= 0) {
				throw new SessionLostException("Broken pipe.");
			} else {
				return recvBytes;
			}
		}
		
		public Message read(MessageBeanFactory factory) {
			int msgSize = 0;
			int msgType = 0;
			int cnt = 0;
			
			if (phase == RecvPhase.MessageSize) {
				if (remaining > 0) {
					cnt = socketRecv(bytes, got, remaining);
					remaining -= cnt;
					got += cnt;
				}

				if (remaining == 0) {
					msgSize = BinaryMessageBeanFormat.GetInt32(bytes, 0);
					Reset(RecvPhase.MessageBody, msgSize);
				}
			} else {
				toRead = Math.Min(remaining, bytes.Length);
				cnt = socketRecv(bytes, 0, toRead);
				recvBuffer.Write(bytes, 0, cnt);
				remaining -= cnt;
						
				if (remaining == 0) {
					recvBuffer.Seek(0, SeekOrigin.Begin);
					msgType = recvBuffer.ReadByte();
					Message message = Message.CreateMessage(msgType, factory);
					message.Deserialize(recvBuffer);
					Reset(RecvPhase.MessageSize, 4);
					
					return message;
				}
			}
			
			return null;
		}
	}

	public class Receiver {
		private Session session;
		private MessageReader reader;
		
		public Receiver(Session session) {
			this.session = session;
			this.reader = new MessageReader();
		}
		
		public void Reset() {
			reader.Reset(session.Socket);
		}

		public void ThreadRun() {
			Socket socket = session.Socket;
			reader.Reset(socket);
			Message message = null;
			bool readable;
			int timeout = session.HeartbeatInterval * 1000000;

			while (true) {
				try {
					readable = socket.Poll(timeout, SelectMode.SelectRead);
					if (readable) {
						message = reader.read(session.MessageBeanFactory);
						if (message != null) {
							session.OnMessage(message);
						}
					} else {
						session.Idle();
					}
				} catch (Exception e) {
					session.onReceiverException(e);
					return;
				}
			}
		}
	}

	public class Session {
		enum State {
			Connecting,		/* 正在建立TCP连接 */
			Requesting,		/* TCP连接已建立，正在申请创建会话 */
			Established,	/* 会话已建立，可以正常收发数据 */
			Closing,		/* 正在关闭连接 */
			Closed			/* 连接已关闭 */
		}

		public enum Error {
			SessionNotClosed		= -1,
			ErrorAddressFormat		= -2,
			ConnectRefused			= -3,
			ConnectionLost			= -4
		}
		
		const int DEFAULT_HEARTBEAT_INTERVAL = 60;
		const int DEFAULT_CONNECT_TIMEOUT = 10;

//		private static Regex ipv4Regex = new Regex("[0-9]+(\\.[0-9]+){3}");
		private State currentState;
		private EndPoint hostEP;
		private Socket socket;
		private UInt64 clientTimestamp, serverTimestamp;
		private string sessionKey;
		private int heartbeatInterval;
		private Sender sender;
		private Receiver receiver;
		private Thread senderThread, receiverThread;
		private Queue<MessageBean> queue;
		private bool checkingHeartbeat;
		private HeartbeatMessage hbREQ, hbACK;
		private MessageBeanFactory factory;
		private bool crypto;
		
		public Socket Socket {
			get { return socket; }
		}
		
		public int HeartbeatInterval {
			get { return heartbeatInterval; }
		}
		
		public MessageBeanFactory MessageBeanFactory {
			get { return factory; }
		}
		
		public bool Established {
			get { return currentState == State.Established; }
		}
		
		public UInt64 ClientTimestamp {
			get { return clientTimestamp; }
		}
		
		public UInt64 ServerTimestamp {
			get { return serverTimestamp; }
		}
		
		public bool Crypto {
			get { return crypto; }
		}
		
		public String SessionKey {
			get { return sessionKey; }
		}
		
		public Session(MessageBeanFactory factory) : this(factory, true) {
		}

		public Session(MessageBeanFactory factory, bool crypto) {
			this.factory = factory;
			this.crypto = crypto;
			
			currentState = State.Closed;
			sender = new Sender(this);
			receiver = new Receiver(this);
			queue = new Queue<MessageBean>();
			socket = null;
			checkingHeartbeat = false;
			hbREQ = new HeartbeatMessage(MessageType.ClientHeartbeat);
			hbACK = new HeartbeatMessage(MessageType.ServerHeartbeatAck);
		}
			
		public void onSenderException(Exception e) {
			currentState = State.Closing;
			Logger.Trace("Sender exception:" + e);
		}
		
		public void onReceiverException(Exception e) {
			currentState = State.Closing;
			Logger.Trace("Receiver exception:" + e);
		}

		public void Close() {
			lock (this) {
				if (socket != null) {
					socket.Close();
					socket = null;
				}

				if (receiverThread != null) {
					if (receiverThread.IsAlive) {
						receiverThread.Interrupt();
					}
					
					receiverThread.Join();
					receiverThread = null;
				}
				
				if (senderThread != null) {
					if (senderThread.IsAlive) {
						senderThread.Interrupt();
					}
					
					senderThread.Join();
					senderThread = null;
				}

				currentState = State.Closed;
			
				queue.Clear();
				checkingHeartbeat = false;
			}
		}
		
		private void StartIOThreads() {
			sender.Reset();
			senderThread = new Thread(new ThreadStart(sender.ThreadRun));
			senderThread.Start();
			
			receiver.Reset();
			receiverThread = new Thread(new ThreadStart(receiver.ThreadRun));
			receiverThread.Start();
		}

		public int Open(string host, int port, int heartbeatInterval, int connectingTimeout) {
			if (currentState != State.Closed) {
				return (int)Error.SessionNotClosed;
			} else {
				this.heartbeatInterval = heartbeatInterval;
			}
			
			/* 与服务器建立连接 */
			currentState = State.Connecting;
			int errcode = Connect(host, port, connectingTimeout);
			if (errcode != 0) {
				Close();
				return errcode;
			}
			
			/* 把socket设置成为nonblocking模式 */
			socket.Blocking = false;
			/* 启动网络IO线程 */
			StartIOThreads();
			
			/* 向服务器申请建立应用层会话 */
			currentState = State.Requesting;
			errcode = RequestSession(connectingTimeout);
			if (errcode != 0) {
				Close();
				return errcode;
			}

			return 0;
		}
		
		public int Open(string host, int port) {
			return Open(host, port, DEFAULT_HEARTBEAT_INTERVAL, DEFAULT_CONNECT_TIMEOUT);
		}

		private int RequestSession(int timeout) {
			try {
				clientTimestamp = Message.CurrentTimeMillis();
				RequestSessionMessage rsm = new RequestSessionMessage(clientTimestamp);
				sender.Send(rsm);
			
				for (int i = 0; i < timeout; ++i) {
					Thread.Sleep(1000);
					if (currentState == State.Established) {
						break;
					}
				}
			} catch (Exception e) {
				Logger.Trace(e);
			}

			if (currentState != State.Established) {
				return (int)Error.ConnectionLost;
			}

			return 0;
		}

		private int Connect(string host, int port, int timeout) {
			try {
//				IPAddress ipAddress;
//				Match m = ipv4Regex.Match(host);
//				if (m.Success) { /* host 是127.0.0.1这样格式的ip地址 */
//					ipAddress = IPAddress.Parse(host);
//					hostEP = new IPEndPoint(ipAddress, port);
//				} else { /* host是域名 */
//					hostEP = new DnsEndPoint(host, port);
//				}
//				
				IPAddress ipAddress = IPAddress.Parse(host);
				hostEP = new IPEndPoint(ipAddress, port);
			} catch (Exception e) {
				Logger.Trace(e);
				return (int)Error.ErrorAddressFormat;
			}
			
			socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
			IAsyncResult result = socket.BeginConnect(hostEP, null, null);
			result.AsyncWaitHandle.WaitOne(timeout * 1000, true);
			return socket.Connected ? 0 : (int)Error.ConnectRefused;

		}
		
		public void Idle() {
			if (checkingHeartbeat) { /* 两次检测心跳失败 */
				Logger.Trace("Check heartbeat failed, close session ...");
				Close();
			} else {
				Logger.Trace("Receiver IDLE, checking heartbeat ...");
				checkingHeartbeat = true;
				sender.Send(hbREQ);
			}
		}
		
		public void OnMessage(Message message) {
			checkingHeartbeat = false;
			
			switch (message.Type) {
			case MessageType.ClientHeartbeatAck:
				Logger.Trace("Received client heartbeat ACK.");
				break;

			case MessageType.ServerHeartbeat:
				sender.Send(hbACK);
				break;

			case MessageType.App:
				CommandMessage cm = (CommandMessage)message;
				lock (this) {
					queue.Enqueue(cm.MessageBean);
				}
				break;
				
			case MessageType.SessionEstablished:
				SessionEstablishedMessage sem = (SessionEstablishedMessage)message;
				serverTimestamp = sem.EstablishedTimestamp;
				sessionKey = sem.SessionKey;
				currentState = State.Established;
				Logger.Trace("Session established, key={0} ts={1}",
						sessionKey, serverTimestamp);
				break;

			default:
				break;
			}
		}

		public void SendMessageBean(MessageBean mbean) {
			if (currentState != State.Established) {
				throw new SessionLostException();
			} else if (senderThread == null) {
				throw new SessionLostException();
			} else if (!senderThread.IsAlive) {
				throw new SessionLostException();
			}

			sender.Send(new CommandMessage(mbean, factory));
		}

		public MessageBean RecvMessageBean() {
			if (currentState != State.Established) {
				throw new SessionLostException();
			} else if (receiverThread == null) {
				throw new SessionLostException();
			} else if (!receiverThread.IsAlive) {
				throw new SessionLostException();
			}
			
			lock (this) {
				return queue.Count > 0 ? queue.Dequeue() : null;
			}
		}
	}
	
	public class SessionLostException : ApplicationException {
		public SessionLostException(string msg) : base(msg) {
		}
		
		public SessionLostException() : base() {
		}
	}
	
	class KeyData {
		public static UInt64[] fibonacci = {
			1L, 1L, 2L, 3L, 5L, 8L, 13L, 21L, 34L, 55L, 89L, 144L, 233L, 377L, 610L,
			987L, 1597L, 2584L, 4181L, 6765L, 10946L, 17711L, 28657L, 46368L,
			75025L, 121393L, 196418L, 317811L, 514229L, 832040L, 1346269L,
			2178309L, 3524578L, 5702887L, 9227465L, 14930352L, 24157817L,
			39088169L, 63245986L, 102334155L, 165580141L, 267914296L,
			433494437L, 701408733L, 1134903170L, 1836311903L, 2971215073L,
			4807526976L, 7778742049L, 12586269025L, 20365011074L,
			32951280099L, 53316291173L, 86267571272L, 139583862445L,
			225851433717L, 365435296162L, 591286729879L, 956722026041L,
			1548008755920L, 2504730781961L, 4052739537881L, 6557470319842L,
			10610209857723L
		};
		public static UInt64[] feeds = {
			4937981208836185383L,
			5582529378488325032L,
			1530270151771565451L,
			3389839389802268617L,
			818775917343865025L,
			4422360002079832991L,
			5396506125082888077L,
			6281417156412459484L,
			2601054018453221965L,
			8146976980905694738L,
			2774689136121125327L,
			1203264123036693182L,
			4158712705292413305L,
			153656531190490364L,
			2328875010753340322L,
			4178919223514220765L,
			6435355743284873911L,
			1850048448781994455L,
			5987007434083745714L,
			7883119513198886834L,
			5214731850309646306L,
			2499815067100629642L,
			2150788269842929996L,
			1531336282780221333L,
			5384696777000890623L,
			3712706158071964194L,
			4077430404767663090L,
			8974960420246845553L,
			4709363160524745260L,
			5658209737268877137L,
			3927175898897456514L,
			5169331952561806602L,
			2289676234970920279L,
			4860670293070175091L,
			630629172882050453L,
			8550751592134673888L,
			1967803798714115396L,
			9177238019227519424L,
			8959084045314876757L,
			1192558918818816706L,
			6541164482600990833L,
			8350222176083562042L,
			5329346925472226006L,
			7462855413572505183L,
			2874177387501827050L,
			6588998375311920177L,
			5341025630127332626L,
			7818368633505310317L,
			5569739050283895999L,
			4426781808004331972L,
			7761780616557105876L,
			2235348271858402836L,
			7755108432203967293L,
			4699801399825338024L,
			5977848668530516018L,
			2224789287101123520L,
			4644720200811464253L,
			8950880294880880157L,
			6509788513075202470L,
			713150281238026022L,
			602714455966728308L,
			1435721900387331108L,
			8007869604350802915L,
			1467671376469862967L,
			3621544852800921645L,
			7580722802752572609L,
			5583244098974674756L,
			993363269019231637L,
			6078960400750617734L,
			3036294486464643398L,
			5778987073689188375L,
			5946057807641163374L,
			3745154839283294037L,
			6703313533609535653L,
			3603492822327807192L,
			4213322765817908698L,
			4900392761625162795L,
			7825021426497354831L,
			8744796820810632519L,
			3410345650705719630L,
			1930262988496825389L,
			5281611571136180830L,
			784946537026600477L,
			1032724589967179436L,
			8536584606747731928L,
			690087618782632812L,
			4654081076882609005L,
			6869618704114777644L,
			2820220775339503101L,
			6296136214517072391L,
			7742248165892014555L,
			4908225718740269686L,
			8666852732861629381L,
			774269286881117024L,
			8387026716629301597L,
			1369174087789566634L,
			1425446471422270997L,
			218320295728824935L,
			4063860053584079208L,
			6221336619816559775L,
			8156532171782076439L,
			3758707807515753928L,
			7741594115902422050L,
			2229692419329642247L,
			4374379216461817997L,
			8407207198806666862L,
			7932612562058652314L,
			7423646229950382319L,
			2283708041001558011L,
			989284346686619718L,
			7215895692631372907L,
			2126126719712902846L,
			1159064796975389229L,
			6833950130022168423L,
			1926032447930993643L,
			1370126640219461189L,
			8706252952800521358L,
			7404228934045691030L,
			3533153106010161104L,
			6136048233101781734L,
			2602403680438438255L,
			7054710716670404301L,
			8937728857767520177L,
			6222237956990542143L,
			1127578845389015675L,
			635537597329469767L,
			2353102507111716565L,
			1676259964530090994L,
		};
	}
}
